package com.markhsiu.minimq.remote.transport.talentaio.handler;

import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.markhsiu.minimq.client.CallBackCache;
import com.markhsiu.minimq.client.CallBackor;
import com.markhsiu.minimq.message.Message;
import com.markhsiu.minimq.message.Result;
import com.markhsiu.minimq.message.constant.MessageCmdEnum;
import com.markhsiu.minimq.remote.transport.talentaio.MessagePacket;
import com.markhsiu.minimq.serialize.SerializeCodec;
import com.markhsiu.minimq.serialize.protostuff.ProtostuffCodec;
import com.talent.aio.client.intf.ClientAioHandler;
import com.talent.aio.common.Aio;
import com.talent.aio.common.ChannelContext;

public class MessageClientAioHandler extends MessageAbsAioHandler
		implements ClientAioHandler<Object, MessagePacket, Object> {

	private static Logger logger = LoggerFactory.getLogger(MessageClientAioHandler.class);
	private static boolean isDebugEnabled = logger.isDebugEnabled();
	
	private static final AtomicInteger count = new AtomicInteger(1);

	private CallBackor callBackor;

	public MessageClientAioHandler(CallBackor callBackor) {
		this.callBackor = callBackor;
	}

	/**
	 * 处理消息
	 */
	@Override
	public Object handler(MessagePacket packet, ChannelContext<Object, MessagePacket, Object> channel)
			throws Exception {

		byte[] body = packet.getBody();
		if (body != null) {
			
			
			SerializeCodec<Message> codec = ProtostuffCodec.OneInstance();
			Message message = codec.deserialize(body);
			if(isDebugEnabled){
				logger.debug(" count = {} ",count.getAndIncrement() );
				logger.debug("收到消息：{}" , message);
			}
			
			Result result = CallBackCache.getCallBack(message.getMessageID());

			if (result != null) {
				result.setMessage(message);
				result.release();
			}

			if (callBackor != null) {

				if (message.getCmd() == MessageCmdEnum.NEW) {
					result = new Result();
					result.setMessage(message);
					Message response = callBackor.print(result);
					MessagePacket resppacket = new MessagePacket();
					byte[] data = codec.serialize(response);
					resppacket.setBody(data);
					Aio.send(channel, resppacket);
				}
			}

		}

		return null;
	}

	private static MessagePacket heartbeatPacket = new MessagePacket();

	/**
	 * 此方法如果返回null，框架层面则不会发心跳；如果返回非null，框架层面会定时发本方法返回的消息包
	 */
	@Override
	public MessagePacket heartbeatPacket() {
		return heartbeatPacket;
	}
}
